package com.zsd.android.rxjavalib;

/**
 * 被观察者
 */
public class Observable<T> {

    private ObservableOnSubscribe source;

    private Observable(ObservableOnSubscribe source) {
        this.source = source;
    }

    /**
     * 参数中：ObservableOnSubscribe<? extends T> 是上限的关系
     * @param source
     * @param <T> 静态方法声明的泛型，和参数中ObservableOnSubscribe<? extends T>的 T 是同一个 T
     * @return
     */
    public static <T> Observable<T> create(ObservableOnSubscribe<? extends T> source) {
        return new Observable<T>(source);
    }

    /**
     *
     * @param ts
     * @param <T>
     * @return
     */
    public static <T> Observable<T> just(final T ... ts) {
        //想办法让source不能为空, 而我们create操作符是使用用自己传进来的
        return new Observable<T>(new ObservableOnSubscribe<T>() {
            @Override
            public void subscribe(Observer<? super T> observableEmitter) {
                for (T t : ts) {
                    //发射用户传递参数数据
                    observableEmitter.onNext(t);
                }
                //调用完毕, 发射完毕事件
                observableEmitter.onComplete();
            }
        });
    }

    /**
     *
     * @param t
     * @param <T>
     * @return
     */
    public static <T> Observable<T> just(final T t) {
        return new Observable<T>(new ObservableOnSubscribe<T>() {
            @Override
            public void subscribe(Observer<? super T> observableEmitter) {
                observableEmitter.onNext(t);
                observableEmitter.onComplete();
            }
        });
    }

    /**
     * ? super T 代表可写
     * ? extends R 代表可读
     * @param function
     * @param <R>
     * @return
     */
    public <R> Observable<R> map(Function<? super T, ? extends R> function) {
        ObservableMap<T, R> observableMap = new ObservableMap(source, function);
        return new Observable<R>(observableMap);
    }

    /**
     * 给上游分配线程
     * @return
     */
    public Observable<T> subscribeOn() {
        return create(new SubscribeOn(source));
    }

    public Observable<T> observerOn() {
        return create(new ObserverOn(source));
    }

    /**
     * 参数中：Observer<? extends T> 是上限的关系
     * @param observer
     */
    public void subscribe(Observer<? extends T> observer) {
        observer.onSubscribe();
        source.subscribe(observer);
    }
}
